"""
Case Type   : 发布订阅冲突解决
Case Name   : 双向订阅，均设置subscription_conflict_resolution=keep_local，数据冲突时保留本地数据
Create At   : 2024/04/10
Owner       : @songjing20
Description :
    1.两集群均设置subscription_conflict_resolution=keep_local
    2.在两个集群创建表
    3.两个集群创建发布订阅
    4.集群1表中插入数据
    5.集群2查询表数据是否同步，插入是两条数据
    6.集群1查询数据是否同步
    7.两个集群开启事务向表中插入数据，主键冲突，先后提交事务
    8.两个集群查询表数据
    9.清理环境
Expect      :
    1.成功
    2.成功
    3.成功
    4.成功
    5.同步，成功
    6.4条数据
    7.成功
    8.集群1保留本地(5,c15)，集群2保留本地数据(5,c25)
    9.成功
History     :
"""

import os
import time
import unittest

from yat.test import macro
from yat.test import Node
from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Common import Common
from testcase.utils.Constant import Constant
from testcase.utils.ComThread import ComThread

Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 > Primary_SH.get_node_num(), '非1+2以上环境不执行')
class Pubsubclass(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f"-----{os.path.basename(__file__)} start-----")
        self.pri_userdb_pub = Node(node='PrimaryDbUser')
        self.pri_userdb_sub = Node(node='remote1_PrimaryDbUser')
        self.constant = Constant()
        self.commsh_pub = CommonSH('PrimaryDbUser')
        self.commsh_sub = CommonSH('remote1_PrimaryDbUser')
        self.com_pub = Common()
        self.com_sub = Common('remote1_PrimaryDbUser')
        self.tb_name = 't_pubsub_0337'
        self.pubname = 'pub_pubsub_0337'
        self.subname = 'sub_pubsub_0337'
        self.conflict_param = 'subscription_conflict_resolution'
        self.pub_port = str(int(self.pri_userdb_pub.db_port) + 1)
        self.sub_port = str(int(self.pri_userdb_sub.db_port) + 1)
        self.pub_wal_level = self.com_pub.show_param("wal_level")
        self.sub_wal_level = self.com_sub.show_param("wal_level",
                                                     macro.DB_ENV_PATH_REMOTE1)
        self.user_param_pub = f'-U {self.pri_userdb_pub.db_user} ' \
                              f'-W {self.pri_userdb_pub.db_password}'
        self.user_param_sub = f'-U {self.pri_userdb_sub.db_user} ' \
                              f'-W {self.pri_userdb_sub.db_password}'
        text = '-----查询集群状态，修改pg_hba，发布端设置wal_level=logical-----'
        self.log.info(text)
        self.log.info('----- 集群1状态 -----')
        status = self.commsh_pub.get_db_cluster_status('detail')
        self.log.info(status)
        self.assertEqual(status.count('Normal'), 4, f'执行失败:{text}')
        self.log.info('----- 集群2状态 -----')
        status = self.commsh_sub.get_db_cluster_status(
            'detail', env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(status)
        self.assertEqual(status.count('Normal'), 4, f'执行失败:{text}')
        guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host  replication  {self.pri_userdb_sub.db_user}  '
            f'{self.pri_userdb_sub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, f'集群1配置流复制权限失败:{text}')
        if 'logical' != self.pub_wal_level:
            res = self.commsh_pub.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=logical')
            self.assertTrue(res, f'集群1设置wal_level失败:{text}')
            restart_res = self.commsh_pub.restart_db_cluster()
            self.assertTrue(restart_res, f'重启集群1失败:{text}')
        if 'logical' != self.sub_wal_level:
            res = self.commsh_sub.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=logical',
                dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
                env_path=macro.DB_ENV_PATH_REMOTE1)
            self.assertTrue(res, f'集群2设置wal_level失败:{text}')
            restart_res = self.commsh_sub.restart_db_cluster(
                env_path=macro.DB_ENV_PATH_REMOTE1)
            self.assertTrue(restart_res, f'重启集群2失败:{text}')
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host  replication  {self.pri_userdb_pub.db_user}  '
            f'{self.pri_userdb_pub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(guc_res, f'集群2配置流复制权限失败:{text}')

    def test_pubsub(self):
        text = '-----step1:设置subscription_conflict_resolution;expect:成功-----'
        self.log.info(text)
        res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            f"{self.conflict_param}='keep_local'")
        self.log.info(res)
        self.assertTrue(res, f'集群1执行失败:{text}')
        res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            f"{self.conflict_param}='keep_local'", single=True,
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(res)
        self.assertTrue(res, f'集群2执行失败:{text}')

        text = '-----step2:在两个集群创建表;expect:成功-----'
        self.log.info(text)
        sql = f"drop table if exists {self.tb_name} cascade;" \
              f"create table {self.tb_name}(" \
              f"c_int int primary key, c_varchar varchar(15));"
        pub_res = self.commsh_pub.execut_db_sql(sql, self.user_param_pub)
        self.log.info(pub_res)
        self.assertEqual(pub_res.count(self.constant.CREATE_TABLE_SUCCESS), 2,
                         f'发布端建表失败:{text}')
        sub_res = self.commsh_sub.execut_db_sql(
            sql, self.user_param_sub, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_res)
        self.assertEqual(pub_res.count(self.constant.CREATE_TABLE_SUCCESS), 2,
                         f'订阅端建表失败:{text}')

        text = '-----step3:两集群创建发布订阅;expect:成功-----'
        self.log.info(text)
        create_pub = f"drop publication if exists {self.pubname};" \
                     f"create publication {self.pubname} " \
                     f"for table {self.tb_name};"
        pub_res = self.commsh_pub.execut_db_sql(create_pub, self.user_param_pub)
        self.log.info(pub_res)
        self.assertIn(self.constant.create_pub_succ_msg, pub_res,
                      f'执行失败:{text}')
        guc_res = self.commsh_sub.execute_generate(
            macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn(self.constant.create_keycipher_success, guc_res,
                      f'执行失败:{text}')
        create_sub = f"drop subscription if exists {self.subname} cascade;" \
                     f"create subscription {self.subname} connection " \
                     f"'host={self.pri_userdb_pub.db_host} " \
                     f"port={self.pub_port} " \
                     f"user={self.pri_userdb_pub.db_user} " \
                     f"dbname={self.pri_userdb_pub.db_name} " \
                     f"password={self.pri_userdb_pub.ssh_password}' " \
                     f"publication {self.pubname};"
        sub_res = self.commsh_sub.execut_db_sql(
            create_sub, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_res)
        self.assertIn(self.constant.create_sub_succ_msg, sub_res,
                      f'创建订阅失败:{text}')

        pub_res = self.commsh_sub.execut_db_sql(
            create_pub, self.user_param_pub, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(pub_res)
        self.assertIn(self.constant.create_pub_succ_msg, pub_res,
                      f'执行失败:{text}')
        guc_res = self.commsh_pub.execute_generate(macro.COMMON_PASSWD)
        self.assertIn(self.constant.create_keycipher_success, guc_res,
                      f'执行失败:{text}')
        create_sub = f"drop subscription if exists {self.subname} cascade;" \
                     f"create subscription {self.subname} connection " \
                     f"'host={self.pri_userdb_sub.db_host} " \
                     f"port={self.sub_port} " \
                     f"user={self.pri_userdb_sub.db_user} " \
                     f"dbname={self.pri_userdb_sub.db_name} " \
                     f"password={self.pri_userdb_sub.ssh_password}' " \
                     f"publication {self.pubname};"
        sub_res = self.commsh_pub.execut_db_sql(create_sub, self.user_param_sub)
        self.log.info(sub_res)
        self.assertIn(self.constant.create_sub_succ_msg, sub_res,
                      f'创建订阅失败:{text}')

        text = '-----step4:集群1表中插入数据;expect:成功-----'
        self.log.info(text)
        insert_res = self.commsh_pub.execut_db_sql(
            f"insert into {self.tb_name} values(1, 'c1'),(4, 'cluster1');",
            self.user_param_pub)
        self.log.info(insert_res)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, insert_res,
                      f'执行失败:{text}')

        text = '-----step5:集群2查询是否同步，插入两条数据;expect:同步，成功-----'
        self.log.info(text)
        self.commsh_pub.check_sub_consistency(
            self.subname, self.user_param_pub, 10)
        sql = f"select * from {self.tb_name};" \
              f"insert into {self.tb_name} values(2, 'c2');" \
              f"insert into {self.tb_name} values(3, 'cluster2');"
        sql_res = self.commsh_sub.execut_db_sql(
            sql, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sql_res)
        self.assertIn('1 | c1\n     4 | cluster1\n(2 rows)', sql_res,
                      f'执行失败:{text}')
        self.assertEqual(sql_res.count(self.constant.INSERT_SUCCESS_MSG), 2,
                         f'执行失败:{text}')

        text = '-----step6:集群1查询表数据;expect:4条数据-----'
        self.log.info(text)
        time.sleep(10)
        select_sql = f"select * from {self.tb_name} order by 1;"
        select_res = self.commsh_pub.execut_db_sql(select_sql,
                                                   self.user_param_sub)
        self.log.info(select_res)
        self.assertIn('1 | c1\n     2 | c2\n     3 | cluster2\n     '
                      '4 | cluster1\n(4 rows)', select_res, f'执行失败:{text}')

        text = '-----step7:两个集群开启事务向表中插入数据，主键冲突，先后提交事务;expect:成功-----'
        self.log.info(text)
        insert1 = ComThread(self.commsh_pub.execut_db_sql, args=(
            f"start transaction;insert into {self.tb_name} values(5, 'c15');"
            f"select pg_sleep(3);commit;",))
        insert1.setDaemon(True)
        insert1.start()
        time.sleep(1)
        insert2_res = self.commsh_sub.execut_db_sql(
            f"start transaction;insert into {self.tb_name} values(5, 'c25');"
            f"select pg_sleep(3);end;", self.user_param_sub, None,
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(insert2_res)
        self.assertTrue(self.constant.INSERT_SUCCESS_MSG in insert2_res and
                        self.constant.COMMIT_SUCCESS_MSG in insert2_res,
                        f'执行失败:{text}')
        insert1.join(10)
        insert1_res = insert1.get_result()
        self.log.info(insert1_res)
        self.assertIn(self.constant.COMMIT_SUCCESS_MSG, insert1_res,
                      f'执行失败:{text}')

        text = '-----step8:两个集群查询表数据;' \
               'expect:集群1保留本地(5,c15)，集群2保留本地数据(5,c15)-----'
        self.log.info(text)
        time.sleep(10)
        res1 = self.commsh_pub.execut_db_sql(select_sql, self.user_param_pub)
        self.log.info(res1)
        res2 = self.commsh_sub.execut_db_sql(
            select_sql, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(res2)
        self.assertIn('5 | c15\n(5 rows)', res1, f'执行失败:{text}')
        self.assertIn('5 | c25\n(5 rows)', res2, f'执行失败:{text}')

    def tearDown(self):
        text = '-----step9:清理环境;expect:成功-----'
        self.log.info(text)
        drop_sql = f"drop table if exists {self.tb_name} cascade;" \
                   f"drop publication if exists {self.pubname};" \
                   f"drop subscription if exists {self.subname};"
        sub_res = self.commsh_sub.execut_db_sql(
            drop_sql, self.user_param_sub, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_res)
        pub_res = self.commsh_pub.execut_db_sql(drop_sql, self.user_param_pub)
        self.log.info(pub_res)
        pub_wal_level_res = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'wal_level={self.pub_wal_level}')
        self.log.info(pub_wal_level_res)
        pub_restart_res = self.commsh_pub.restart_db_cluster()
        self.log.info(pub_restart_res)
        sub_wal_level_res = self.commsh_sub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'wal_level={self.sub_wal_level}',
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_wal_level_res)
        sub_restart_res = self.commsh_pub.restart_db_cluster()
        self.log.info(sub_restart_res)
        pub_guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_userdb_sub.db_user} '
            f'{self.pri_userdb_sub.db_host}/32 ')
        self.log.info(pub_guc_res)
        sub_guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host    replication  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_guc_res)
        res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            f"{self.conflict_param}='error'", single=True,
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(res)
        self.assertTrue(res, f'执行失败:{text}')
        self.assertTrue('ERROR' not in sub_res and 'WARNING' not in sub_res,
                        f'执行失败:{text}')
        self.assertTrue('ERROR' not in sub_res and 'WARNING' not in pub_res,
                        f'执行失败:{text}')
        self.assertTrue(pub_wal_level_res, f'集群1执行失败:{text}')
        self.assertTrue(sub_wal_level_res, f'集群2执行失败:{text}')
        self.assertTrue(pub_restart_res, f'重启集群1失败:{text}')
        self.assertTrue(sub_restart_res, f'重启集群2失败:{text}')
        self.assertTrue(pub_guc_res, f'执行失败:{text}')
        self.assertTrue(sub_guc_res, f'执行失败:{text}')
        self.log.info(f"-----{os.path.basename(__file__)} end-----")
